package com.xxk.rocket.star.manage;

import com.alibaba.rocketmq.client.consumer.MQPushConsumer;
import com.xxk.rocket.star.core.consume.ConsumeFactory;
import com.xxk.rocket.star.core.consume.ConsumeHandlerImpl;
import com.xxk.rocket.star.core.exception.StargateException;

import java.util.Map;

public class StargateConsumeManageImpl implements IStargateConsumeManage {

  @Override
  public boolean startAll() {
    Map<String, MQPushConsumer> concurrentMap = ConsumeFactory.getConcurrentMap();
    for (String key : concurrentMap.keySet()) {
      boolean b = ConsumeFactory.startMQConsumer(key);
      if (!b) {
        throw new StargateException("startAll error group:" + key);
      }
    }
    return true;
  }

  @Override
  public void closeAll() {
    new Thread() {
      @Override
      public void run() {
        Map<String, MQPushConsumer> concurrentMap = ConsumeFactory.getConcurrentMap();
        for (MQPushConsumer mqPushConsumer : concurrentMap.values()) {
          mqPushConsumer.shutdown();
        }
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        for (MQPushConsumer mqPushConsumer : concurrentMap.values()) {
          ConsumeFactory.getMessageListenerByConsumer(mqPushConsumer).getConsumeHandler().shutdown();
        }
      }
    }.start();
  }

  @Override
  public boolean isAllClose() {
    Map<String, MQPushConsumer> concurrentMap = ConsumeFactory.getConcurrentMap();
    for (MQPushConsumer mqPushConsumer : concurrentMap.values()) {
      ConsumeHandlerImpl handler = ConsumeFactory.getMessageListenerByConsumer(mqPushConsumer).getConsumeHandler();
      if (!handler.isShutdown()) {
        return false;
      }
    }
    return true;
  }
}
